package cn.itcast.nems.listener;

import cn.hutool.core.exceptions.ExceptionUtil;
import cn.hutool.core.lang.Assert;
import cn.itcast.nems.approval.bpm.eventstrategy.BpmEventStrategy;
import cn.itcast.nems.approval.dao.entity.ApprovalForm;
import cn.itcast.nems.approval.service.ApprovalFormService;
import cn.itcast.nems.approval.service.ApprovalFormService.JournalType;
import cn.itcast.nems.common.account.AccountHolder;
import cn.itcast.nems.common.account.AccountInfo;
import cn.itcast.nems.common.tenant.TenantInfo;
import cn.itcast.nems.common.tenant.TenantInfoHolder;
import cn.itcast.nems.manager.msas.MsasManager;
import cn.itcast.nems.manager.redis.RedisUtil;
import cn.itcast.nems.manager.support.ApprovalMessageHolder;
import cn.itcast.nems.manager.support.ApprovalMessageInfo;
import cn.itcast.nems.manager.util.WeixinMessageUtil;
import cn.itcast.nems.manager.weixin.WeixinManager;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.itcast.activiti.sdk.domain.StartProcessInstance;
import com.itcast.activiti.sdk.domain.TaskStart;
import com.itcast.msas.sdk.acm.domain.AccountDto;
import jakarta.annotation.PostConstruct;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * 监听并处理iBPM MQ消息（审批流程发起、审批人审批、知会、审批流程结束）
 */
@Slf4j
@Component
@RocketMQMessageListener(
        topic = "${bpm.rocketmq.topic}",
        consumerGroup = "${bpm.rocketmq.consumer.git_task_apply}",
        selectorExpression = "TAG_TASK_APPROVE")
@RequiredArgsConstructor
public class BpmWorkflowEventListener implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    public static final String EVENTNAME_START = "start";
    public static final String EVENTNAME_END = "end";
    public static final String EVENTNAME_TAKE = "take";
    public static final String TENANT_ID = "NEMS";
    @Value("${spring.profiles.active}")
    private String ACTIVE;
    @Autowired
    private ApprovalFormService approvalFormService;
    // 事件处理策略类集合
    @Autowired
    private Set<BpmEventStrategy> eventStrategySets;
    // 通过凭单类型快速获取策略类
    private Map<String, BpmEventStrategy> eventStrategyMaps;
    @Autowired
    private MsasManager msasManager;
    @Autowired
    private WeixinManager weixinManager;
    @Autowired
    private RedisUtil redisUtil;

    private final String REDIS_BPM_JOURNAL_ID_PREFIX = "NEMS:BPM:JOURNAL:ID:";
    private final String REDIS_BPM_MESSAGE_ID_PREFIX = "NEMS:BPM:MESSAGE:ID:";

    @PostConstruct
    public void init() {
        eventStrategyMaps = new HashMap<>();
        for (BpmEventStrategy bpmEventStrategy : eventStrategySets) {
            eventStrategyMaps.put(bpmEventStrategy.getStrategyName(), bpmEventStrategy);
        }
    }

    @Override
    public void onMessage(MessageExt messageExt) {
        try {
            log.info("【BPM消息监听】msgId：{} ，消息体：{} ", messageExt.getMsgId(), JSON.toJSONString(messageExt));
            String payload = new String(messageExt.getBody(), StandardCharsets.UTF_8);
            JSONObject message = JSONObject.parseObject(payload);
            log.info("TaskApproveQueueListener {}", message);
            TenantInfoHolder.setTenantInfo(TenantInfo.DEFAULT_TENANT_INFO);
            // 只接收新EMS系统的审批事件
            String tenantId = message.getString(TaskStart.Fields.tenantId);
            if (!TENANT_ID.equalsIgnoreCase(tenantId)) {
                return;
            }
            String passbackParam = message.getString(StartProcessInstance.Fields.passbackParam);
            if (StringUtils.hasText(passbackParam)) {
                if (!ACTIVE.equalsIgnoreCase(passbackParam)) {
                    return;
                }
            }
            // 通过 businessKey 获取对应审批单信息，也能进一步确定凭单类型
            String businessKey = message.getString(StartProcessInstance.Fields.businessKey);
            Assert.notBlank(businessKey, "审批事件报文找不到businessKey！");

            ApprovalForm approvalForm = approvalFormService.getByBusinessKey(businessKey);
            if (approvalForm == null) {
                log.info("审批事件中包含的businessKey[{}]找不到对应的审批单！", businessKey);
                return;
            }
            this.setAccount(approvalForm);
            // 休学、待定班、同学科转班 则都使用相同的策略处理类
            String eventStrategyKey = approvalForm.getJournalType();
            if (JournalType.EXCHANGE.name().equalsIgnoreCase(eventStrategyKey) ||
                    JournalType.DELAY.name().equalsIgnoreCase(eventStrategyKey) ||
                    JournalType.SUSPENSION.name().equalsIgnoreCase(eventStrategyKey)) {
                eventStrategyKey = JournalType.TRANSFER.name();
            }

            // 找到不同凭单对应的BPM事件处理类
            BpmEventStrategy eventStrategy = eventStrategyMaps.get(eventStrategyKey);
            if (Objects.isNull(eventStrategy)) {
                log.error("凭单[{}]找不到对应的处理类！", approvalForm.getJournalType());
                return;
            }
            final String event = message.getString("event");
            final String msgId = messageExt.getMsgId();

            JSONObject taskRequestParam = message.getJSONObject("taskRequestParam");
            Boolean audit = taskRequestParam.getBoolean(TaskStart.Fields.audit);
            if (EVENTNAME_START.equals(event)) {
                // 开始事件不处理
            } else if (EVENTNAME_TAKE.equals(event)) {
                // 需要处理抄送人
                // 1：审批）2：抄送）
                String category = message.getString("category");
                if ("2".equals(category)) { // 只需要处理抄送事件，其他事件在接口调用时已处理
                    String redisKey = REDIS_BPM_MESSAGE_ID_PREFIX + msgId;
                    if (redisUtil.hasKey(redisKey)) {
                        log.info("【BPM消息监听】审批 msgId：{} ，已成功消费过，消息体：{} ", messageExt.getMsgId(), JSON.toJSONString(messageExt));
                        return;
                    }
                    String assignees = message.getString("assignee");
                    for (String targetAssignee : Arrays.asList(assignees.split(","))) {
                        eventStrategy.inform(businessKey, targetAssignee);
                    }
                    redisUtil.set(redisKey, msgId, 60 * 60);
                }
                if (message.containsKey("backState") && "驳回至发起人".equals(message.getString("backState"))) {
                    // 驳回发起人后凭单进入草稿状态
                    eventStrategy.rollbackRequester(businessKey);
                    // 驳回至发起人
                    approvalFormService.rollbackRequester(businessKey);
                    return;
                } else if (message.containsKey("backState") && "驳回至上一级".equals(message.getString("backState"))) {
                    // 驳回至上一级
                    approvalFormService.rollbackPrevious(businessKey);
                }
            } else if (EVENTNAME_END.equals(event)) {
                // 需要处理抄送人
                // 1：审批）2：抄送）
                String category = message.getString("category");
                if ("2".equals(category)) { // 只需要处理抄送事件，其他事件在接口调用时已处理
                    String assignees = message.getString("assignee");
                    for (String targetAssignee : Arrays.asList(assignees.split(","))) {
                        eventStrategy.inform(businessKey, targetAssignee);
                    }
                }
                if (audit) {
                    try {
                        final String journalType = approvalForm.getJournalType();
                        final JournalType referenceType = JournalType.valueOf(journalType);
                        ApprovalMessageHolder.set(new ApprovalMessageInfo(approvalForm.getId(),
                                approvalForm.getJournalId(),
                                referenceType.getCnName(),
                                approvalForm.getBpmBizKey(),
                                this.ACTIVE,
                                msgId));
                        String redisKey = REDIS_BPM_JOURNAL_ID_PREFIX + businessKey;
                        if (redisUtil.hasKey(redisKey)) {
                            log.info("【BPM消息监听】审批流结束 msgId：{} ，已成功消费过，消息体：{} ", messageExt.getMsgId(), JSON.toJSONString(messageExt));
                            return;
                        }
                        final Runnable callback = eventStrategy.finish(businessKey);
                        if (callback != null) {
                            callback.run();
                        }
                        redisUtil.set(redisKey, msgId, 60 * 60 * 3);
                    } catch (Exception e) {
                        this.processFinishException(approvalForm, msgId, e);
                        throw e;
                    } finally {
                        ApprovalMessageHolder.clear();
                    }
                }
            } else {
                log.warn("审批消息：{} 不支持的处理类型", message);
            }
            log.info("消息{}被{}成功处理", msgId, this.getHostIp());
        } catch (Exception ex) {
            log.warn(ex.getMessage(), ex);
            throw ex;
        } finally {
            AccountHolder.clear();
            TenantInfoHolder.clear();
        }
    }

    private void setAccount(ApprovalForm approvalForm) {
        final String email = approvalForm.getCreatedBy();
        final AccountDto account = this.msasManager.getAccount(email);
        final AccountInfo accountInfo = new AccountInfo(email,
                account.getName(),
                account.getEmployeeNumber(),
                "");
        AccountHolder.setAccountInfo(accountInfo);
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 最长消费时间24小时
        defaultMQPushConsumer.setConsumeTimeout(1 * 24 * 60);
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 最大重试次数小于等于 15 次
        defaultMQPushConsumer.setMaxReconsumeTimes(15);
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
    }

    private void processFinishException(ApprovalForm approvalForm, String msgId, Exception e) {
        String businessKey = approvalForm.getBpmBizKey();
        final String journalId = approvalForm.getJournalId();
        log.warn("处理凭单结束出现错误，审批单ID: {}, 凭单ID: {}, 消息ID: {} 异常信息: {}", businessKey, journalId, msgId, e.getMessage());
        final String message = this.createWeixinMessage(approvalForm, msgId, e);
        this.weixinManager.sendMarkdownMessage(msgId, message);
        /*
        try {
            eventStrategy.finishFail(businessKey, e);
        } catch (Exception ex) {
            log.warn(ex.getMessage(), ex);
        }
         */
        String msg = ExceptionUtil.getRootCauseMessage(e);
        approvalFormService.updateDescription(approvalForm.getId(), msg);
    }

    private String createWeixinMessage(ApprovalForm approvalForm, String msgId, Exception e) {
        final String journalType = approvalForm.getJournalType();
        final JournalType referenceType = JournalType.valueOf(journalType);
        final AccountInfo acc = AccountHolder.getRequiredAccountInfo();
        return WeixinMessageUtil.createWeixinMessage(approvalForm.getJournalId(),
                referenceType.getCnName(),
                approvalForm.getBpmBizKey(),
                msgId,
                acc.getName(),
                acc.getEmail(),
                this.ACTIVE,
                e);
    }

    private String getHostIp() {
        String ipAddress = "Unknown";
        try {
            InetAddress localHost = InetAddress.getLocalHost();
            ipAddress = localHost.getHostAddress();
        } catch (UnknownHostException e) {
            log.warn(e.getMessage(), e);
        }
        return ipAddress;
    }
}